package com.cetccloud.ap.proxy.coord;

import com.alibaba.fastjson.JSON;
import com.cetccloud.ap.proxy.common.CacheSink;
import com.cetccloud.ap.proxy.common.ProxyContext;
import com.cetccloud.ap.proxy.common.event.CallbackEvent;
import io.vertx.core.Vertx;
import io.vertx.core.eventbus.EventBus;
import io.vertx.core.eventbus.MessageConsumer;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.cache.CacheManager;
import reactor.core.publisher.MonoSink;

import javax.annotation.PostConstruct;

/**
 * @ClassName AbstractCoord
 * @Description TODO
 * @Author superz
 * @Date 2021/4/11 10:43
 * @Version 1.0
 **/
@Slf4j
public abstract class AbstractCoord implements Coord {

    @Autowired
    protected Vertx vertx;

    @Autowired
    protected CacheManager cacheManager;

    @PostConstruct
    private void init() {
        buildMessageConsumer(vertx.eventBus());
    }

    private void buildMessageConsumer(EventBus eventBus) {
        MessageConsumer<String> consumer = eventBus.consumer(eventbusAddress());
        consumer.handler(message -> {
            CallbackEvent event = JSON.parseObject(message.body(), CallbackEvent.class);
            CacheSink cacheSink = cacheManager.getCache(cacheName()).get(event.getIdentity(), CacheSink.class);
            if (null == cacheSink) {
                log.info("回调sink不存在: {}", event.getIdentity());
            } else {
                log.info("回调处理: {}", event.getIdentity());
                cacheManager.getCache(cacheName()).evict(event.getIdentity());
                dealCallback(event, cacheSink.getMonoSink(), cacheSink.getProxyContext());
            }
        });
    }

    protected abstract void dealCallback(CallbackEvent event, MonoSink sink, ProxyContext context);

    @Override
    public void registerSink(String identity, MonoSink sink, ProxyContext context) {
        cacheManager.getCache(cacheName()).put(identity, CacheSink.builder().monoSink(sink).proxyContext(context).build());
    }

    @Override
    public void success(String identity, Object data) {
        // topic模式
        vertx.eventBus().publish(eventbusAddress(), JSON.toJSONString(new CallbackEvent(identity, data, null)));
    }

    @Override
    public void error(String identity, String message) {
        // topic模式
        vertx.eventBus().publish(eventbusAddress(), JSON.toJSONString(new CallbackEvent(identity, null, message)));
    }

    protected abstract String cacheName();

    protected abstract String eventbusAddress();
}
